package com.myorg.service;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.util.LineReader;
import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.message.BasicNameValuePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.gson.Gson;
import com.myorg.domain.BlockInfo;
import com.myorg.domain.Email;
import com.myorg.domain.EmailDelimiter;
import com.myorg.domain.SearchRequest;
import com.myorg.domain.SearchResult;
import com.myorg.search.condition.SearchCondition;
import com.myorg.util.CompressionHelper;
import com.myorg.util.ConstantsToBeExternalized;
import com.myorg.util.DataLocalityRunner;
import com.myorg.util.DataLocalityRunner.DataLocalityTask;
import com.myorg.util.HdfsHelper;
import com.myorg.util.StreamFeedToDecompressor;
import com.myorg.util.Utility;

// TODO: Auto-wiring is not there yet.
public class SearchService extends ScanService {
	private static final Logger logger = LoggerFactory.getLogger(SearchService.class);
	private final static Text EOL = new Text("\n");

	public SearchResult dispatchSearch(final SearchRequest request, final String contextPath, final int port, final String syncPath) {
		RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(ConstantsToBeExternalized.DEFAULT_SOCKET_TIMEOUT).setConnectTimeout(ConstantsToBeExternalized.DEFAULT_CONNECTION_TIMEOUT)
				.build();
		final SearchResult rtn = new SearchResult();
		final CloseableHttpAsyncClient httpclient = HttpAsyncClients.custom().setDefaultRequestConfig(requestConfig).setMaxConnTotal(Integer.MAX_VALUE).setMaxConnPerRoute(Integer.MAX_VALUE).build();
		try {
			List<BlockInfo> blockInfos = generateBlockInfoList(request.getFilePaths());
			dumpBlockInfo(blockInfos);

			httpclient.start();
			final CountDownLatch latch = new CountDownLatch(blockInfos.size());
			DataLocalityRunner.runWithLocality(blockInfos, new DataLocalityTask() {
				@Override
				public void run(BlockInfo blkInfo, String selectedHost) {
					try {
						URI uri = new URI("http", null, selectedHost, port, contextPath + "/DoSearch", "start=" + blkInfo.getBlockLocation().getOffset() + "&end="
								+ (blkInfo.getBlockLocation().getLength() + blkInfo.getBlockLocation().getOffset() - 1) + "&max=" + request.getMax() + "&compress="
								+ (blkInfo.isFileCompressed() ? "Y" : "N") + "&syncPath=" + syncPath, null);
						logger.info("Dispatching to URL: " + uri);

						final HttpPost httpPost = new HttpPost(uri);
						List<NameValuePair> params = new ArrayList<NameValuePair>(2);
						params.add(new BasicNameValuePair("source", blkInfo.getFilePath()));
						params.add(new BasicNameValuePair("key", request.getKey()));
						List<String> ignoreIdList = request.getIgnoreIds();
						if (ignoreIdList != null) {
							for (String id : ignoreIdList) {
								params.add(new BasicNameValuePair("ignoreId", id));
							}
						}
						httpPost.setEntity(new UrlEncodedFormEntity(params, "UTF-8"));

						httpclient.execute(httpPost, new FutureCallback<HttpResponse>() {
							public void completed(final HttpResponse resp) {
								// TODO handle socket time out.

								logger.info(httpPost.getRequestLine() + "->" + resp.getStatusLine());
								String responseStr = Utility.readHttpResponse(resp);
								SearchResult searchResult = new Gson().fromJson(responseStr, SearchResult.class);
								rtn.merge(searchResult);
								latch.countDown();
							}

							public void failed(final Exception ex) {
								latch.countDown();
								// TODO handle socket timeout here, maybe retry or send the request to another host.
								logger.info(httpPost.getRequestLine() + "->" + ex);
							}

							public void cancelled() {
								latch.countDown();
								logger.info(httpPost.getRequestLine() + " cancelled");
							}
						});
					} catch (URISyntaxException e) {
						e.printStackTrace();
						throw new RuntimeException(e);
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
						throw new RuntimeException(e);
					}
				}
			});

			latch.await();
			logger.info("completed");

			return rtn.max(request.getMax());
		} catch (InterruptedException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			try {
				httpclient.close();
			} catch (IOException ee) {
			}
		}
	}

	private List<Email> searchCompressedText(SearchCondition searchCondition, long startL, long endL, FSDataInputStream fileIn, int max, List<String> ignoreIdList) throws IOException {
		// StringBuilder builder = new StringBuilder();
		List<Email> rtn = new ArrayList<Email>();
		long pos;
		// long length = endL - startL;
		BufferedInputStream bis = new BufferedInputStream(fileIn, ConstantsToBeExternalized.BUFFER_SIZE_FOR_DECOMPRESSION);
		PushbackInputStream pis = new PushbackInputStream(bis, ConstantsToBeExternalized.BUFFER_SIZE_FOR_DECOMPRESSION);

		boolean isThe3Bytes = CompressionHelper.isGZ3byte(pis);
		if (isThe3Bytes) {
			pos = startL;
		} else {
			pos = startL + CompressionHelper.readUptoNextGZ3byte(pis);
		}

		while (pos <= endL && rtn.size() < max) {
			Text value = new Text();
			Text emailId = new Text();
			int offset = readNextForCompressedText(pis, value, emailId);

			if (!ignoreIdList.contains(emailId.toString())) {
				Email email = new Email().setId(emailId.toString()).setContent(value.toString());
				String summary = email.matchAndSummarize(searchCondition);
				if (summary != null) {
					rtn.add(email.setSummary(summary));
				}
			}

			// No byte read, seems that we reached end of Split
			// Break and return false (no key / value)
			if (offset == 0) {
				break;
			}
			// Line is read, new position is set
			pos += offset;
		}

		return rtn;
	}

	private int max(long... maxs) {
		int rtn = ConstantsToBeExternalized.MAX_LENGTH_OF_EMAIL;
		if (maxs != null && maxs.length > 0) {
			for (long m : maxs) {
				if (m < rtn) {
					rtn = (int) m;
				}
			}
		}
		return rtn;
	}

	private List<Email> searchPlainText(SearchCondition searchCondition, long startL, long endL, FSDataInputStream fileIn, int max, List<String> ignoreIdList) throws IOException {
		List<Email> rtn = new ArrayList<Email>();

		long pos;
		LineReader in = new LineReader(fileIn, ConstantsToBeExternalized.DEFAULT_HDFS_READ_BUFFER_SIZE);
		long length = endL - startL + 1;

		// If Split "S" starts at byte 0, first line will be processed
		// If Split "S" does not start at byte 0, first line has been already
		// processed by "S-1" and therefore needs to be silently ignored
		boolean skipFirstLine = false;
		if (startL != 0) {
			skipFirstLine = true;
			// Set the file pointer at "start - 1" position.
			// This is to make sure we won't miss any line
			// It could happen if "start" is located on a EOL
			--startL;
			fileIn.seek(startL);
		}

		// If first line needs to be skipped, read first line
		// and stores its content to a dummy Text
		if (skipFirstLine) {
			Text dummy = new Text();
			// Reset "start" to "start + line offset"
			startL += in.readLine(dummy, 0, max(length));
		}
		pos = startL;

		while (pos <= endL && rtn.size() < max) {
			Text value = new Text();
			Text emailId = new Text();
			// Read first line and store its content to "value"
			int offset = readNext(in, value, emailId);

			if (!ignoreIdList.contains(emailId.toString())) {
				Email email = new Email().setId(emailId.toString()).setContent(value.toString());
				String summary = email.matchAndSummarize(searchCondition);
				if (summary != null) {
					rtn.add(email.setSummary(summary));
				}
			}

			// No byte read, seems that we reached end of Split
			// Break and return false (no key / value)
			if (offset == 0) {
				break;
			}
			// Line is read, new position is set
			pos += offset;
		}

		in.close();
		return rtn;
	}

	private int readNext(LineReader in, Text text, Text emailId) throws IOException {
		int offset = 0;
		Text tmp = new Text();
		boolean doLoop = true;

		while (doLoop) {
			int offsetTmp = in.readLine(tmp);
			offset += offsetTmp;
			// End of File
			if (offsetTmp == 0) {
				break;
			}
			String eId = EmailDelimiter.getEmailIdIfDelimitor(tmp.toString());
			if (eId != null) {
				emailId.set(eId);
			}
			doLoop ^= eId != null;

			// Append value to record
			if (doLoop) {
				text.append(tmp.getBytes(), 0, tmp.getLength());
				text.append(EOL.getBytes(), 0, EOL.getLength());
			}
		}

		return offset;
	}

	private int readNextForCompressedText(PushbackInputStream pis, Text text, Text emailId) throws IOException {
		int offset = 0;
		StreamFeedToDecompressor std = null;
		boolean doLoop = true;

		while (doLoop) {
			std = CompressionHelper.parse(pis);
			InputStream gIs = std.getIn();
			StringBuilder textbuilder = CompressionHelper.decompress(gIs);
			gIs.close();
			long offsetTmp = std.getNumByteRead();

			offset += offsetTmp;
			// End of File
			if (offsetTmp == 0) {
				break;
			}

			String eId = EmailDelimiter.getEmailIdIfDelimitor(textbuilder.toString());
			if (eId != null) {
				emailId.set(eId);
			}
			doLoop ^= eId != null;

			// Append value to record
			if (doLoop) {
				String tmp = textbuilder.toString();
				text.append(tmp.getBytes(), 0, tmp.length());
				text.append(EOL.getBytes(), 0, EOL.getLength());
			}
		}

		return offset;
	}

	public SearchResult doSearch(final String source, final String startS, final String endS, String key, final int max, final List<String> ignoreIdList, final boolean compress) {
		logger.info("---------------Searh Start ----------------------------------");
		logger.info("source=" + source);
		logger.info("startS=" + startS);
		logger.info("endS=" + endS);
		logger.info("key=" + key);
		logger.info("max=" + max);
		logger.info("ignoreIdList=" + ignoreIdList);
		logger.info("compress=" + compress);

		final SearchCondition searchCondition = SearchCondition.parse(key);

		if (searchCondition != null) {
			Future<SearchResult> future = pool.submit(new Callable<SearchResult>() {
				@Override
				public SearchResult call() throws Exception {
					final SearchResult rtn = new SearchResult();
					Utility.profiling(new Runnable() {
						@Override
						public void run() {
							try {
								HdfsHelper helper = new HdfsHelper();
								long startL = Long.valueOf(startS);
								long endL = Long.valueOf(endS);
								FileSystem hdfs = FileSystem.newInstance(helper.getConfig());
								Path srcPath = new Path(source);
								FSDataInputStream fileIn = hdfs.open(srcPath);
								if (compress) {
									rtn.setList(searchCompressedText(searchCondition, startL, endL, fileIn, max, ignoreIdList));
								} else {
									rtn.setList(searchPlainText(searchCondition, startL, endL, fileIn, max, ignoreIdList));
								}

								fileIn.close();
								hdfs.close();
							} catch (IOException e) {
								e.printStackTrace();
								throw new RuntimeException(e);
							}
						}
					}, ">>>>>>>>> Search on Data Node: ");
					return rtn;
				}
			});

			try {
				return future.get();
			} catch (InterruptedException e) {
				e.printStackTrace();
				throw new RuntimeException(e);
			} catch (ExecutionException e) {
				e.printStackTrace();
				throw new RuntimeException(e);
			}
		} else {
			logger.info("searchCondition is null with key = " + key);
			return new SearchResult();
		}
	}
}
